package cn.doitedu.sql;

import beans.Person;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/5
 * @Desc: 学大数据，上多易教育
 *   流 转 表
 **/
public class _23_StreamToTable_Demo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);
        // 1,a,18
        SingleOutputStreamOperator<JavaBean> mappedStream = stream.map(new MapFunction<String, JavaBean>() {
            @Override
            public JavaBean map(String value) throws Exception {
                String[] split = value.split(",");

                return new JavaBean(Integer.parseInt(split[0]), split[1], Long.parseLong(split[2]),Long.parseLong(split[3]));
            }
        });

        // 把流转成表对象，如果流中的数据类型是javaBean，那么在转表时底层会自动推断出表结构
        Table table = tenv.fromDataStream(mappedStream);   // 得到 table api的表对象，后续可以写table api来查询计算
        // tenv.fromChangelogStream(mappedStream);   // 这个方法的特点是，把一个changelog流转成 表对象

        // 把流转成带名字的“表”（视图），后续可以写sql来查询计算
        tenv.createTemporaryView("table_tmp",mappedStream);  // 自动推断表结构 schema
        tenv.executeSql("desc table_tmp").print();  // 可以从打印结果中，看到底层已经进行了表结构的自动推断
        /**
         *+------+--------+-------+-----+--------+-----------+
         * | name |   type |  null | key | extras | watermark |
         * +------+--------+-------+-----+--------+-----------+
         * |   id |    INT | FALSE |     |        |           |
         * | name | STRING |  TRUE |     |        |           |
         * |  age | BIGINT | FALSE |     |        |           |
         * +------+--------+-------+-----+--------+-----------+
         */


        /**
         *
         * create table table_tmp2(
         *    id int not null,
         *    name string not null,
         *    age bigint,
         *    ts bigint,
         *    name_2 as upper(name),
         *    rt as to_timestamp_ltz(ts,3),
         *    watermark for rt as rt - interval '0' second,
         *    primary key (id,name)
         * )
         *
         */
        // 手动指定表结构
        tenv.createTemporaryView("table_tmp2",mappedStream,
                Schema.newBuilder()
                        .column("id","int not null")  // 类型可以用字符串表达
                        .column("name","string not null")
                        .column("age",DataTypes.BIGINT())  // 类型也可以用api来表达
                        .column("ts",DataTypes.BIGINT())
                        .columnByExpression("name_2","upper(name)")
                        .columnByExpression("rt","to_timestamp_ltz(ts,3)")
                        .watermark("rt","rt - interval '0' second ")
                        .primaryKey("id","name")
                        .build()

        );

        tenv.executeSql("desc table_tmp2").print();





        env.execute();
    }

    @NoArgsConstructor
    @AllArgsConstructor
    @Data
    public static class JavaBean{
        private int id;
        private String name;
        private long age;
        private long ts;
    }



}

